package com.atguigu.datastream.day05;

import com.atguigu.datastream.bean.Event;
import com.atguigu.datastream.test.day03.Flink_05_Source_useDefault;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * ClassName: Flink07_ProcessTimeWindow_AggFunWithProcessWindowFun
 * Package: com.atguigu.day05
 * Description:
 *
 *             在网站的各种统计指标中，一个很重要的统计指标就是热门的链接；想要得到热门的url，前提是得到每个链接的“热门度”。一般情况下，可以用url的浏览量（点击量）表示热门度。我们这里统计10秒钟的url浏览量，每5秒钟更新一次；另外为了更加清晰地展示，还应该把窗口的起始结束时间一起输出。
 *
 * @Author ChenJun
 * @Create 2023/4/11 18:19
 * @Version 1.0
 */
public class Flink07_ProcessTimeWindow_AggFunWithProcessWindowFun {
    public static void main(String[] args) throws Exception {

        //1. 创建流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2. 通过自定义数据源获取数据
        DataStreamSource<Event> streamSource = env.addSource(new Flink_05_Source_useDefault.ClickSource());

        //3.将相同url的数据聚合达到同一个分区
        KeyedStream<Event, String> keyedStream = streamSource.keyBy(new KeySelector<Event, String>() {
            @Override
            public String getKey(Event event) throws Exception {
                return event.url;
            }
        });

        //4. 开启一个滑动窗口，每5秒钟计算一次，计算10秒的数据
        WindowedStream<Event, String, TimeWindow> window = keyedStream.window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)));

        //5.先使用增量聚合函数得到这个页面的点击量，然后再利用全窗口函数，打上窗口的起始和结束时间
        window.aggregate(new AggregateFunction<Event, Integer, Integer>() {
            @Override
            public Integer createAccumulator() {
                return 0;
            }

            @Override
            public Integer add(Event event, Integer integer) {
                return integer + 1;
            }

            @Override
            public Integer getResult(Integer integer) {
                return integer;
            }

            @Override
            public Integer merge(Integer a, Integer b) {
                return a + b;
            }
        }, new ProcessWindowFunction<Integer, String, String, TimeWindow>() {
            @Override
            public void process(String s, ProcessWindowFunction<Integer, String, String, TimeWindow>.Context context, Iterable<Integer> iterable, Collector<String> collector) throws Exception {

                Integer count = 0;

                for (Integer integer : iterable) {
                    count = integer;
                }

                collector.collect("[" + context.window().getStart() + "," + context.window().getEnd() + ")" + "热门度为：" + count + "数据个数为：" + iterable.spliterator().estimateSize());

            }
        }).print();

        env.execute();

    }
}
